package com.baishancloud.log.test

import cn.hutool.json.JSONUtil
import com.baishancloud.log.common.entity._
import com.baishancloud.log.common.env.StreamEnv
import com.baishancloud.log.common.source.{BroadcastUtil, SourceUtil}
import com.baishancloud.log.common.util.ExternalDataFetchUtil
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.datastream.BroadcastStream
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

import scala.collection.mutable
import scala.util.control.Breaks

/**
 * 测试获取单个资源的广播流
 *
 * @author ziqiang.wang 
 * @date 2022/2/18 14:28
 */
object TestCacheData extends Serializable {

  def main(args: Array[String]): Unit = {
    val parameterTool: ParameterTool = ParameterTool.fromArgs(args)
    println(s"输入参数：${parameterTool.toMap.toString}")
    StreamEnv.builder(parameterTool)
    val source: DataStream[Long] = SourceUtil.numberEmit()
    val domainInfoSource: BroadcastStream[mutable.Map[String, DomainEntity]] = BroadcastUtil.domainInfo()
    val accountDomainInfoSource: BroadcastStream[mutable.Map[String, DomainEntity]] = BroadcastUtil.accountDomainInfo(accounts = "mfniu,adesk_20200629_del")
    val serverCacheGroupInfoSource: BroadcastStream[mutable.Map[String, ServerCacheGroupEntity]] = BroadcastUtil.serverCacheGroupInfo()
    val countryInfoSource: BroadcastStream[mutable.Map[String, CountryEntity]] = BroadcastUtil.countryInfo()
    val provinceInfoSource: BroadcastStream[mutable.Map[String, ProvinceEntity]] = BroadcastUtil.provinceInfo()
    val ispInfoSource: BroadcastStream[mutable.Map[String, IspEntity]] = BroadcastUtil.ispInfo()
    val userInfoSource: BroadcastStream[mutable.Map[String, Int]] = BroadcastUtil.userInfo()
    val nodeInfoSource: BroadcastStream[mutable.Map[String, NodeEntity]] = BroadcastUtil.nodeInfo()
    val typeModuleInfoSource: BroadcastStream[mutable.Map[String, mutable.Set[Int]]] = BroadcastUtil.typeModuleInfo()
    val serviceTypeInfoSource: BroadcastStream[mutable.Map[String, Byte]] = BroadcastUtil.serviceTypeInfo()
    val svgInfoSource: BroadcastStream[mutable.Map[String, String]] = BroadcastUtil.svgInfo()

    source
      .connect(domainInfoSource)
      .process(new BroadcastProcessFunction[Long, mutable.Map[String, DomainEntity], DomainEntity] {
        var domainInfo: mutable.Map[String, DomainEntity] = mutable.Map[String, DomainEntity]()
        val LOGGER: Logger = LoggerFactory.getLogger("domainInfo")
        var subtask: Int = -1

        override def open(parameters: Configuration): Unit = {
          subtask = getRuntimeContext.getIndexOfThisSubtask
          ExternalDataFetchUtil.domainInfoFetchFromFile(domainInfo)
          LOGGER.info(s"并行度编号：${subtask}，domainInfo初始化完成，数据量：${domainInfo.size}，数据示例：\n${outStr()}")
        }

        override def processElement(value: Long, ctx: BroadcastProcessFunction[Long, mutable.Map[String, DomainEntity], DomainEntity]#ReadOnlyContext, out: Collector[DomainEntity]): Unit = {

        }

        override def processBroadcastElement(value: mutable.Map[String, DomainEntity], ctx: BroadcastProcessFunction[Long, mutable.Map[String, DomainEntity], DomainEntity]#Context, out: Collector[DomainEntity]): Unit = {
          if (value.nonEmpty) {
            domainInfo = value
            LOGGER.info(s"并行度编号：${subtask}，domainInfo接收到广播数据，数据量：${domainInfo.size}，数据示例：\n${outStr()}")
          } else {
            LOGGER.error("广播流发生了异常")
          }
        }

        private def outStr(): String = {
          var str: String = ""
          var count: Int = 0
          val iterator: Iterator[(String, DomainEntity)] = domainInfo.iterator
          Breaks.breakable {
            while (count < 10) {
              val tuple = iterator.next()
              str += (tuple._1 + "->" + JSONUtil.toJsonStr(tuple._2) + "\n")
              count += 1
            }
          }
          str
        }
      })
      .print()

    source
      .connect(accountDomainInfoSource)
      .process(new BroadcastProcessFunction[Long, mutable.Map[String, DomainEntity], DomainEntity] {
        var accountDomainInfo: mutable.Map[String, DomainEntity] = mutable.Map[String, DomainEntity]()
        val LOGGER: Logger = LoggerFactory.getLogger("accountDomainInfo")
        var subtask: Int = -1

        override def open(parameters: Configuration): Unit = {
          subtask = getRuntimeContext.getIndexOfThisSubtask
          ExternalDataFetchUtil.accountDomainInfoFetchFromFile("mfniu,adesk_20200629_del", accountDomainInfo)
          LOGGER.info(s"并行度编号：${subtask}，accountDomainInfo初始化完成，数据量：${accountDomainInfo.size}，数据示例：\n${outStr()}")
        }

        override def processElement(value: Long, ctx: BroadcastProcessFunction[Long, mutable.Map[String, DomainEntity], DomainEntity]#ReadOnlyContext, out: Collector[DomainEntity]): Unit = {

        }

        override def processBroadcastElement(value: mutable.Map[String, DomainEntity], ctx: BroadcastProcessFunction[Long, mutable.Map[String, DomainEntity], DomainEntity]#Context, out: Collector[DomainEntity]): Unit = {
          if (value.nonEmpty) {
            accountDomainInfo = value
            LOGGER.info(s"并行度编号：${subtask}，accountDomainInfo收到广播数据，数据量：${accountDomainInfo.size}，数据示例：\n${outStr()}")
          } else {
            LOGGER.error("广播流发生了异常")
          }
        }

        private def outStr(): String = {
          var str: String = ""
          var count: Int = 0
          val iterator: Iterator[(String, DomainEntity)] = accountDomainInfo.iterator
          Breaks.breakable {
            while (iterator.hasNext && count < 10) {
              val tuple = iterator.next()
              str += (tuple._1 + "->" + JSONUtil.toJsonStr(tuple._2) + "\n")
              count += 1
            }
          }
          str
        }
      })
      .print()


    source
      .connect(serverCacheGroupInfoSource)
      .process(new BroadcastProcessFunction[Long, mutable.Map[String, ServerCacheGroupEntity], ServerCacheGroupEntity] {
        var serverCacheGroupInfo: mutable.Map[String, ServerCacheGroupEntity] = mutable.Map[String, ServerCacheGroupEntity]()
        val LOGGER: Logger = LoggerFactory.getLogger("ipCacheGroupInfo")
        var subtask: Int = -1

        override def open(parameters: Configuration): Unit = {
          subtask = getRuntimeContext.getIndexOfThisSubtask
          ExternalDataFetchUtil.serverCacheGroupFetchFromFile(serverCacheGroupInfo)
          LOGGER.info(s"并行度编号：${subtask}，serverCacheGroupInfo初始化完成，数据量：${serverCacheGroupInfo.size}，数据示例：\n${outStr()}")
        }

        override def processElement(value: Long, ctx: BroadcastProcessFunction[Long, mutable.Map[String, ServerCacheGroupEntity], ServerCacheGroupEntity]#ReadOnlyContext, out: Collector[ServerCacheGroupEntity]): Unit = {

        }

        override def processBroadcastElement(value: mutable.Map[String, ServerCacheGroupEntity], ctx: BroadcastProcessFunction[Long, mutable.Map[String, ServerCacheGroupEntity], ServerCacheGroupEntity]#Context, out: Collector[ServerCacheGroupEntity]): Unit = {
          if (value.nonEmpty) {
            serverCacheGroupInfo = value
            LOGGER.info(s"并行度编号：${subtask}，serverCacheGroupInfo收到广播数据，数据量：${serverCacheGroupInfo.size}，数据示例：\n${outStr()}")
          } else {
            LOGGER.error("广播流发生了异常")
          }
        }

        private def outStr(): String = {
          var str: String = ""
          var count: Int = 0
          val iterator: Iterator[(String, ServerCacheGroupEntity)] = serverCacheGroupInfo.iterator
          Breaks.breakable {
            while (count < 10) {
              val tuple = iterator.next()
              str += (tuple._1 + "->" + JSONUtil.toJsonStr(tuple._2) + "\n")
              count += 1
            }
          }
          str
        }
      })
      .print()

    source
      .connect(countryInfoSource)
      .process(new BroadcastProcessFunction[Long, mutable.Map[String, CountryEntity], CountryEntity] {
        var countryInfo: mutable.Map[String, CountryEntity] = mutable.Map[String, CountryEntity]()
        val LOGGER: Logger = LoggerFactory.getLogger("countryInfo")
        var subtask: Int = -1

        override def open(parameters: Configuration): Unit = {
          subtask = getRuntimeContext.getIndexOfThisSubtask
          ExternalDataFetchUtil.countryInfoFetchFromFile(countryInfo)
          LOGGER.info(s"并行度编号：${subtask}，countryInfo初始化完成，数据量：${countryInfo.size}，数据示例：\n${outStr()}")
        }

        override def processElement(value: Long, ctx: BroadcastProcessFunction[Long, mutable.Map[String, CountryEntity], CountryEntity]#ReadOnlyContext, out: Collector[CountryEntity]): Unit = {

        }

        override def processBroadcastElement(value: mutable.Map[String, CountryEntity], ctx: BroadcastProcessFunction[Long, mutable.Map[String, CountryEntity], CountryEntity]#Context, out: Collector[CountryEntity]): Unit = {
          if (value.nonEmpty) {
            countryInfo = value
            LOGGER.info(s"并行度编号：${subtask}，countryInfo收到广播数据，数据量：${countryInfo.size}，数据示例：\n${outStr()}")
          } else {
            LOGGER.error("广播流发生了异常")
          }
        }

        private def outStr(): String = {
          var str: String = ""
          var count: Int = 0
          val iterator: Iterator[(String, CountryEntity)] = countryInfo.iterator
          Breaks.breakable {
            while (count < 10) {
              val tuple = iterator.next()
              str += (tuple._1 + "->" + JSONUtil.toJsonStr(tuple._2) + "\n")
              count += 1
            }
          }
          str
        }
      })
      .print()

    source
      .connect(provinceInfoSource)
      .process(new BroadcastProcessFunction[Long, mutable.Map[String, ProvinceEntity], ProvinceEntity] {
        var provinceInfo: mutable.Map[String, ProvinceEntity] = mutable.Map[String, ProvinceEntity]()
        val LOGGER: Logger = LoggerFactory.getLogger("provinceInfo")
        var subtask: Int = -1

        override def open(parameters: Configuration): Unit = {
          subtask = getRuntimeContext.getIndexOfThisSubtask
          ExternalDataFetchUtil.provinceInfoFetchFromFile(provinceInfo)
          LOGGER.info(s"并行度编号：${subtask}，provinceInfo初始化完成，数据量：${provinceInfo.size}，数据示例：\n${outStr()}")
        }

        override def processElement(value: Long, ctx: BroadcastProcessFunction[Long, mutable.Map[String, ProvinceEntity], ProvinceEntity]#ReadOnlyContext, out: Collector[ProvinceEntity]): Unit = {

        }

        override def processBroadcastElement(value: mutable.Map[String, ProvinceEntity], ctx: BroadcastProcessFunction[Long, mutable.Map[String, ProvinceEntity], ProvinceEntity]#Context, out: Collector[ProvinceEntity]): Unit = {
          if (value.nonEmpty) {
            provinceInfo = value
            LOGGER.info(s"并行度编号：${subtask}，provinceInfo收到广播数据，数据量：${provinceInfo.size}，数据示例：\n${outStr()}")
          } else {
            LOGGER.error("广播流发生了异常")
          }
        }

        private def outStr(): String = {
          var str: String = ""
          var count: Int = 0
          val iterator: Iterator[(String, ProvinceEntity)] = provinceInfo.iterator
          Breaks.breakable {
            while (count < 10) {
              val tuple = iterator.next()
              str += (tuple._1 + "->" + JSONUtil.toJsonStr(tuple._2) + "\n")
              count += 1
            }
          }
          str
        }
      })
      .print()


    source
      .connect(ispInfoSource)
      .process(new BroadcastProcessFunction[Long, mutable.Map[String, IspEntity], IspEntity] {
        var ispInfo: mutable.Map[String, IspEntity] = mutable.Map[String, IspEntity]()
        val LOGGER: Logger = LoggerFactory.getLogger("ispInfo")
        var subtask: Int = -1

        override def open(parameters: Configuration): Unit = {
          subtask = getRuntimeContext.getIndexOfThisSubtask
          ExternalDataFetchUtil.ispInfoFetchFromFile(ispInfo)
          LOGGER.info(s"并行度编号：${subtask}，并行度编号：${subtask},ispInfo初始化完成，数据量：${ispInfo.size}，数据示例：\n${outStr()}")
        }

        override def processElement(value: Long, ctx: BroadcastProcessFunction[Long, mutable.Map[String, IspEntity], IspEntity]#ReadOnlyContext, out: Collector[IspEntity]): Unit = {

        }

        override def processBroadcastElement(value: mutable.Map[String, IspEntity], ctx: BroadcastProcessFunction[Long, mutable.Map[String, IspEntity], IspEntity]#Context, out: Collector[IspEntity]): Unit = {
          if (value.nonEmpty) {
            ispInfo = value
            LOGGER.info(s"并行度编号：${subtask}，ispInfo收到广播数据，数据量：${ispInfo.size}，数据示例：\n${outStr()}")
          } else {
            LOGGER.error("广播流发生了异常")
          }
        }

        private def outStr(): String = {
          var str: String = ""
          var count: Int = 0
          val iterator: Iterator[(String, IspEntity)] = ispInfo.iterator
          Breaks.breakable {
            while (count < 10) {
              val tuple = iterator.next()
              str += (tuple._1 + "->" + JSONUtil.toJsonStr(tuple._2) + "\n")
              count += 1
            }
          }
          str
        }
      })
      .print()

    source
      .connect(userInfoSource)
      .process(new BroadcastProcessFunction[Long, mutable.Map[String, Int], Int] {
        var userInfo: mutable.Map[String, Int] = mutable.Map[String, Int]()
        val LOGGER: Logger = LoggerFactory.getLogger("userInfo")
        var subtask: Int = -1

        override def open(parameters: Configuration): Unit = {
          subtask = getRuntimeContext.getIndexOfThisSubtask
          ExternalDataFetchUtil.userInfoFetchFromFile(userInfo)
          LOGGER.info(s"并行度编号：${subtask}，userInfo初始化完成，数据量：${userInfo.size}，数据示例：\n${outStr()}")
        }

        override def processElement(value: Long, ctx: BroadcastProcessFunction[Long, mutable.Map[String, Int], Int]#ReadOnlyContext, out: Collector[Int]): Unit = {

        }

        override def processBroadcastElement(value: mutable.Map[String, Int], ctx: BroadcastProcessFunction[Long, mutable.Map[String, Int], Int]#Context, out: Collector[Int]): Unit = {
          if (value.nonEmpty) {
            userInfo = value
            LOGGER.info(s"并行度编号：${subtask}，userInfo收到广播数据，数据量：${userInfo.size}，数据示例：\n${outStr()}")
          } else {
            LOGGER.error("广播流发生了异常")
          }
        }

        private def outStr(): String = {
          var str: String = ""
          var count: Int = 0
          val iterator: Iterator[(String, Int)] = userInfo.iterator
          Breaks.breakable {
            while (count < 10) {
              val tuple = iterator.next()
              str += (tuple._1 + "->" + tuple._2 + "\n")
              count += 1
            }
          }
          str
        }
      })
      .print()

    source
      .connect(nodeInfoSource)
      .process(new BroadcastProcessFunction[Long, mutable.Map[String, NodeEntity], NodeEntity] {
        var nodeInfo: mutable.Map[String, NodeEntity] = mutable.Map[String, NodeEntity]()
        val LOGGER: Logger = LoggerFactory.getLogger("nodeInfo")
        var subtask: Int = -1

        override def open(parameters: Configuration): Unit = {
          subtask = getRuntimeContext.getIndexOfThisSubtask
          ExternalDataFetchUtil.nodeInfoFetchFromFile(nodeInfo)
          LOGGER.info(s"并行度编号：${subtask}，nodeInfo初始化完成，数据量：${nodeInfo.size}，数据示例：\n${outStr()}")
        }

        override def processElement(value: Long, ctx: BroadcastProcessFunction[Long, mutable.Map[String, NodeEntity], NodeEntity]#ReadOnlyContext, out: Collector[NodeEntity]): Unit = {

        }

        override def processBroadcastElement(value: mutable.Map[String, NodeEntity], ctx: BroadcastProcessFunction[Long, mutable.Map[String, NodeEntity], NodeEntity]#Context, out: Collector[NodeEntity]): Unit = {
          if (value.nonEmpty) {
            nodeInfo = value
            LOGGER.info(s"并行度编号：${subtask}，nodeInfo收到广播数据，数据量：${nodeInfo.size}，数据示例：\n${outStr()}")
          } else {
            LOGGER.error("广播流发生了异常")
          }
        }

        private def outStr(): String = {
          var str: String = ""
          var count: Int = 0
          val iterator: Iterator[(String, NodeEntity)] = nodeInfo.iterator
          Breaks.breakable {
            while (count < 10) {
              val tuple = iterator.next()
              str += (tuple._1 + "->" + JSONUtil.toJsonStr(tuple._2) + "\n")
              count += 1
            }
          }
          str
        }
      })
      .print()

    source
      .connect(typeModuleInfoSource)
      .process(new BroadcastProcessFunction[Long, mutable.Map[String, mutable.Set[Int]], mutable.Set[Int]] {
        var typeModuleInfo: mutable.Map[String, mutable.Set[Int]] = mutable.Map[String, mutable.Set[Int]]()
        val LOGGER: Logger = LoggerFactory.getLogger("typeModuleInfo")
        var subtask: Int = -1

        override def open(parameters: Configuration): Unit = {
          subtask = getRuntimeContext.getIndexOfThisSubtask
          ExternalDataFetchUtil.typeModuleInfoFetchFromFile(typeModuleInfo)
          LOGGER.info(s"并行度编号：${subtask}，typeModuleInfo初始化完成，数据量：${typeModuleInfo.size}，数据示例：\n${outStr()}")
        }

        override def processElement(value: Long, ctx: BroadcastProcessFunction[Long, mutable.Map[String, mutable.Set[Int]], mutable.Set[Int]]#ReadOnlyContext, out: Collector[mutable.Set[Int]]): Unit = {

        }

        override def processBroadcastElement(value: mutable.Map[String, mutable.Set[Int]], ctx: BroadcastProcessFunction[Long, mutable.Map[String, mutable.Set[Int]], mutable.Set[Int]]#Context, out: Collector[mutable.Set[Int]]): Unit = {
          if (value.nonEmpty) {
            typeModuleInfo = value
            LOGGER.info(s"并行度编号：${subtask}，typeModuleInfo收到广播数据，数据量：${typeModuleInfo.size}，数据示例：\n${outStr()}")
          } else {
            LOGGER.error("广播流发生了异常")
          }
        }

        private def outStr(): String = {
          var str: String = ""
          var count: Int = 0
          val iterator: Iterator[(String, mutable.Set[Int])] = typeModuleInfo.iterator
          Breaks.breakable {
            while (count < 10) {
              val tuple = iterator.next()
              str += (tuple._1 + "->" + JSONUtil.toJsonStr(tuple._2.toArray) + "\n")
              count += 1
            }
          }
          str
        }
      })
      .print()

    source
      .connect(serviceTypeInfoSource)
      .process(new BroadcastProcessFunction[Long, mutable.Map[String, Byte], Byte] {
        var serviceTypeInfo: mutable.Map[String, Byte] = mutable.Map[String, Byte]()
        val LOGGER: Logger = LoggerFactory.getLogger("serviceTypeInfo")
        var subtask: Int = -1

        override def open(parameters: Configuration): Unit = {
          subtask = getRuntimeContext.getIndexOfThisSubtask
          ExternalDataFetchUtil.serviceTypeInfoFetchFromFile(serviceTypeInfo)
          LOGGER.info(s"并行度编号：${subtask}，serviceTypeInfo初始化完成，数据量：${serviceTypeInfo.size}，数据示例：\n${outStr()}")
        }

        override def processElement(value: Long, ctx: BroadcastProcessFunction[Long, mutable.Map[String, Byte], Byte]#ReadOnlyContext, out: Collector[Byte]): Unit = {

        }

        override def processBroadcastElement(value: mutable.Map[String, Byte], ctx: BroadcastProcessFunction[Long, mutable.Map[String, Byte], Byte]#Context, out: Collector[Byte]): Unit = {
          if (value.nonEmpty) {
            serviceTypeInfo = value
            LOGGER.info(s"并行度编号：${subtask}，serviceTypeInfo收到广播数据，数据量：${serviceTypeInfo.size}，数据示例：\n${outStr()}")
          } else {
            LOGGER.error("广播流发生了异常")
          }
        }

        private def outStr(): String = {
          var str: String = ""
          var count: Int = 0
          val iterator: Iterator[(String, Byte)] = serviceTypeInfo.iterator
          Breaks.breakable {
            while (count < 10) {
              val tuple = iterator.next()
              str += (tuple._1 + "->" + tuple._2 + "\n")
              count += 1
            }
          }
          str
        }
      })
      .print()

    source
      .connect(svgInfoSource)
      .process(new BroadcastProcessFunction[Long, mutable.Map[String, String], Byte] {
        var svgInfo: mutable.Map[String, String] = mutable.Map[String, String]()
        val LOGGER: Logger = LoggerFactory.getLogger("svgInfo")
        var subtask: Int = -1

        override def open(parameters: Configuration): Unit = {
          subtask = getRuntimeContext.getIndexOfThisSubtask
          ExternalDataFetchUtil.svgInfoFetchFromFile(svgInfo)
          LOGGER.info(s"并行度编号：${subtask}，svgInfo初始化完成，数据量：${svgInfo.size}，数据示例：\n${outStr()}")
        }

        override def processElement(value: Long, ctx: BroadcastProcessFunction[Long, mutable.Map[String, String], Byte]#ReadOnlyContext, out: Collector[Byte]): Unit = {

        }

        override def processBroadcastElement(value: mutable.Map[String, String], ctx: BroadcastProcessFunction[Long, mutable.Map[String, String], Byte]#Context, out: Collector[Byte]): Unit = {
          if (value.nonEmpty) {
            svgInfo = value
            LOGGER.info(s"并行度编号：${subtask}，svgInfo收到广播数据，数据量：${svgInfo.size}，数据示例：\n${outStr()}")
          } else {
            LOGGER.error("广播流发生了异常")
          }
        }

        private def outStr(): String = {
          var str: String = ""
          var count: Int = 0
          val iterator: Iterator[(String, String)] = svgInfo.iterator
          Breaks.breakable {
            while (count < 10) {
              val tuple = iterator.next()
              str += (tuple._1 + "->" + tuple._2 + "\n")
              count += 1
            }
          }
          str
        }
      })
      .print()

    StreamEnv.execute(parameterTool)

  }

}
